package avicit.bdp.dds.api.service;

import avicit.bdp.common.base.BaseService;
import avicit.bdp.common.datasource.BaseDataSource;
import avicit.bdp.common.dto.response.dds.ProcessDefinitionDTO;
import avicit.bdp.common.dto.response.dds.TaskInstanceDTO;
import avicit.bdp.common.service.dto.CommonTypeDTO;
import avicit.bdp.common.service.dto.DataSourceDTO;
import avicit.bdp.common.service.service.CommonTypeService;
import avicit.bdp.common.service.service.DataSourceService;
import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.DESUtils;
import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.common.utils.enums.CommonTypeEnums;
import avicit.bdp.common.utils.redis.RedisCacheHelper;
import avicit.bdp.core.poi.util.xml.XmlUtils;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.api.dto.ProcessDefinitionStateDto;
import avicit.bdp.dds.api.enums.ProcessType;
import avicit.bdp.dds.api.enums.Status;
import avicit.bdp.dds.api.utils.BloodAnalyseUtils;
import avicit.bdp.dds.api.utils.CheckUtils;
import avicit.bdp.dds.api.utils.LineageUtils;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.graph.DAG;
import avicit.bdp.dds.dao.entity.*;
import avicit.bdp.dds.dao.mapper.ProcessDefinitionMapper;
import avicit.bdp.dds.dao.mapper.ProcessInstanceMapper;
import avicit.bdp.dds.dao.mapper.ProjectMapper;
import avicit.bdp.dds.dispatch.enums.*;
import avicit.bdp.dds.dispatch.model.ProcessData;
import avicit.bdp.dds.dispatch.model.Server;
import avicit.bdp.dds.dispatch.model.TaskNode;
import avicit.bdp.dds.service.process.ProcessService;
import avicit.bdp.dds.service.quartz.MonitorDirectoryScheduleJob;
import avicit.bdp.dds.service.quartz.ProcessScheduleJob;
import avicit.bdp.dds.service.quartz.QuartzExecutors;
import avicit.platform6.api.system.ConvertColumnClient;
import avicit.platform6.api.system.impl.SystemConstant;
import avicit.platform6.commons.utils.BusinessUtil;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.context.ThreadContextHelper;
import avicit.platform6.core.exception.BusinessException;
import avicit.platform6.core.properties.PlatformConstant;
import avicit.platform6.core.rest.msg.QueryRespBean;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.dialect.mysql.visitor.MySqlSchemaStatVisitor;
import com.alibaba.druid.sql.parser.ParserException;
import com.alibaba.druid.stat.TableStat;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

import javax.servlet.http.HttpServletResponse;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.URLEncoder;
import java.text.MessageFormat;
import java.text.ParseException;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author mayanj
 */
@Service
public class ProcessDefinitionService
        extends BaseService<ProcessDefinitionMapper, ProcessDefinition> {

    private static final Logger logger = LoggerFactory.getLogger(ProcessDefinitionService.class);

    private final SchedulerService schedulerService;

    private final ProcessService processService;

    private final MonitorService monitorService;

    private final ProcessInstanceMapper processInstanceMapper;

    private final TaskInstanceService taskInstanceService;

    private final ProjectMapper projectMapper;

    private final CommonTypeService commonTypeService;

    private final ConvertColumnClient convertColumnClient;

    private final DataSourceService dataSourceService;

    private final ProcessResourceConfigService processResourceConfigService;

    private final ProcessDefinitionHistoryService processDefinitionHistoryService;

    private final ProcessDefinitionTemplateService processDefinitionTemplateService;
    private final RedisCacheHelper redisCacheHelper;
    @Autowired
    private BloodConfigService bloodConfigService;
    @Autowired
    private LineageUtils lineageUtils;
    @Autowired
    private LineageTableService lineageTableService;
    @Autowired
    private LineageFieldService lineageFieldService;
    @Autowired
    ExecutorService executorService;

    @Value(value = "${spring.datasource.url}")
    private String url;

    @Value(value = "${spring.datasource.username}")
    private String username;

    @Value(value = "${spring.datasource.password}")
    private String password;

    @Autowired
    public ProcessDefinitionService(
            ConvertColumnClient convertColumnClient,
            SchedulerService schedulerService,
            ProcessService processService,
            MonitorService monitorService,
            ProcessInstanceMapper processInstanceMapper,
            ProcessDefinitionTemplateService processDefinitionTemplateService,
            TaskInstanceService taskInstanceService,
            ProjectMapper projectMapper,
            CommonTypeService commonTypeService,
            DataSourceService dataSourceService,
            ProcessResourceConfigService processResourceConfigService,
            ProcessDefinitionHistoryService processDefinitionHistoryService,
            RedisCacheHelper redisCacheHelper) {

        this.convertColumnClient = convertColumnClient;
        this.schedulerService = schedulerService;
        this.processService = processService;
        this.monitorService = monitorService;
        this.processInstanceMapper = processInstanceMapper;
        this.processDefinitionTemplateService = processDefinitionTemplateService;
        this.taskInstanceService = taskInstanceService;
        this.projectMapper = projectMapper;
        this.commonTypeService = commonTypeService;
        this.dataSourceService = dataSourceService;
        this.processResourceConfigService = processResourceConfigService;
        this.processDefinitionHistoryService = processDefinitionHistoryService;
        this.redisCacheHelper = redisCacheHelper;
    }

    /**
     * whether the graph has a ring
     *
     * @param taskNodeResponseList task node response list
     * @return if graph has cycle flag
     */
    private synchronized boolean graphHasCycle(List<TaskNode> taskNodeResponseList) {
        DAG<String, TaskNode, String> graph = new DAG<>();

        // Fill the vertices
        for (TaskNode taskNodeResponse : taskNodeResponseList) {
            graph.addNode(taskNodeResponse.getName(), taskNodeResponse);
        }

        // Fill edge relations
        for (TaskNode taskNodeResponse : taskNodeResponseList) {
            List<String> preTasks = JSONUtils.toList(taskNodeResponse.getPreTasks(), String.class);
            if (CollectionUtils.isNotEmpty(preTasks)) {
                for (String preTask : preTasks) {
                    if (!graph.addEdge(preTask, taskNodeResponse.getName())) {
                        return true;
                    }
                }
            }
        }

        return graph.hasCycle();
    }

    /**
     * check the process definition node meets the specifications
     *
     * @param processData process data
     */
    void checkProcessNodeList(ProcessData processData) {
        try {
            if (processData == null) {
                logger.error("process data is null");
                throw new BusinessException("流程数据为空");
            }

            // Check whether the task node is normal
            List<TaskNode> taskNodes = processData.getTasks();
            if (taskNodes == null || taskNodes.isEmpty()) {
                logger.error("process node info is empty");
                throw new BusinessException("必须有一个流程节点");
            }

            // check has cycle
            if (graphHasCycle(taskNodes)) {
                logger.error("process DAG has cycle");
                throw new BusinessException("流程节点形成闭环");
            }

            // check whether the process definition json is normal
            for (TaskNode taskNode : taskNodes) {
                if (!CheckUtils.checkTaskNodeParameters(taskNode.getParams(), taskNode.getType())) {
                    logger.error("task node {} parameter invalid", taskNode.getName());
                    throw new BusinessException(taskNode.getName() + "节点参数错误");
                }

                // check extra params
                CheckUtils.checkOtherParams(taskNode.getExtras());
            }

        } catch (Exception e) {
            e.printStackTrace();
            throw new BusinessException(e.getMessage());
        }
    }

    /**
     * create process definition
     *
     * @param processDefinition process Definition
     */
    public String createProcessDefinition(ProcessDefinition processDefinition) {
        // 采集系统新增的流程定义，流程定义ID是数据采集的jobId
        if (!"datax".equals(processDefinition.getTypeId())) {
            /*
             * 场景1：手动新增时，重新生成id，保证系统唯一；
             * 场景2：导入新增时，不论系统中是否存在id，重新生成id，保证系统唯一
             */
            processDefinition.setId(ComUtil.getId());
        }

        if (StringUtils.isBlank(processDefinition.getTypeId())) {
            processDefinition.setTypeId("-1");
        }

        if (StringUtils.isNotBlank(processDefinition.getProcessTemplateId())) {
            DataSourceDTO dataSourceDTO =
                    dataSourceService.selectByPrimaryKey(ProcessDefinition.TEMPLATE_DATASOURCE_ID);
            if (dataSourceDTO == null) {
                addTypeAndDataSource();
            }

            ProcessDefinitionTemplate processDefinitionTemplate =
                    processDefinitionTemplateService.selectByPrimaryKey(
                            processDefinition.getProcessTemplateId());
            processDefinition.setProcessDefinitionJson(
                    processDefinitionTemplate.getProcessDefinitionJson());
            processDefinition.setConnects(processDefinitionTemplate.getConnects());
            processDefinition.setLocations(processDefinitionTemplate.getLocations());
        }

        if (processDefinition.getType() == ProcessType.DATA_SYNC.getCode()
                || processDefinition.getType() == ProcessType.DATA_TRANSFER.getCode()) {
            // step1:检查JSON串合法性
            String processDefinitionJson = checkProcessNodeJson(processDefinition);

            // step2:设置流程定义参数
            if (processDefinition.getReleaseState() == null) {
                processDefinition.setReleaseState(ReleaseState.OFFLINE.getCode());
            }
            processDefinition.setProcessDefinitionJson(processDefinitionJson);
        }

        // 解析数据血缘
        handleLineageRelation(processDefinition);

        // step3:流程定义持久化
        insert(processDefinition);

        // 记录日志
        BdpLogUtil.log4Insert(processDefinition);
        return processDefinition.getId();
    }

    private void handleBloodRelation(ProcessDefinition processDefinition) {
        List<BloodConfigDTO> bloodList = BloodAnalyseUtils.bloodAnalyse(processDefinition);
        BloodConfigDTO query = new BloodConfigDTO();
        query.setProcessDefinitionId(processDefinition.getId());
        bloodConfigService.delete(query);
        // TODO 待实现批量插入
        for (BloodConfigDTO blood : bloodList) {
            bloodConfigService.insert(blood);
        }
    }

    /**
     * 解析数据血缘
     * @param processDefinition
     */
    private void handleLineageRelation(ProcessDefinition processDefinition) {
        Map<String, Object> map = lineageUtils.lineageAnalyse(processDefinition);
        if(map.get("lineageTableList") != null){
            List<LineageTableDTO> lineageTableList = (List<LineageTableDTO>) map.get("lineageTableList");
            LineageTableDTO lineageTable = new LineageTableDTO();
            lineageTable.setProcessDefinitionId(processDefinition.getId());
            lineageTableService.delete(lineageTable);
            // 批量插入
            lineageTableService.batchInsert(lineageTableList);
        }

        if(map.get("lineageFieldList") != null){
            List<LineageFieldDTO> lineageFieldList = (List<LineageFieldDTO>) map.get("lineageFieldList");
            LineageFieldDTO lineageField = new LineageFieldDTO();
            lineageField.setProcessDefinitionId(processDefinition.getId());
            lineageFieldService.delete(lineageField);
            // 批量插入
            lineageFieldService.batchInsert(lineageFieldList);
        }
    }

    private void addTypeAndDataSource() {
        CommonTypeDTO commonTypeDTO = new CommonTypeDTO();
        commonTypeDTO.setId(ProcessDefinition.TEMPLATE_TYPE_ID);
        commonTypeDTO.setName(ProcessDefinition.TEMPLATE_TYPE_NAME);
        commonTypeDTO.setUnfolded(0);
        commonTypeDTO.setLeaf(1);
        commonTypeDTO.setType("dataSourceType");
        commonTypeDTO.setIcon("database");
        commonTypeDTO.setOrderBy(101);
        commonTypeDTO.setParentId("0");
        commonTypeService.insert(commonTypeDTO);

        DataSourceDTO dataSourceDTO = new DataSourceDTO();
        dataSourceDTO.setId(ProcessDefinition.TEMPLATE_DATASOURCE_ID);
        dataSourceDTO.setDatasourceName(ProcessDefinition.TEMPLATE_DATASOURCE_NAME);
        dataSourceDTO.setDatasourceType("mysql");
        dataSourceDTO.setTypeId(ProcessDefinition.TEMPLATE_TYPE_ID);

        String ip = "";
        String port = "";
        Pattern p = Pattern.compile("(\\d+\\.\\d+\\.\\d+\\.\\d+)\\:(\\d+)");
        Matcher m = p.matcher(url);
        while (m.find()) {
            ip = m.group(1);
            port = m.group(2);
        }
        Map<String, String> paramMap = new HashMap<>(16);
        paramMap.put("user", username);
        paramMap.put("password", password);
        paramMap.put("ip", ip);
        paramMap.put("port", port);
        paramMap.put("database", "bdp_demodata");
        paramMap.put("type", "mysql");
        String param = JSON.toJSONString(paramMap);

        dataSourceDTO.setConnectionParams(param);
        dataSourceDTO.setProjectType(0);
        dataSourceDTO.setDatasourceStructure(0);
        BaseDataSource baseDataSource = DBUtils.getBaseDataSource(dataSourceDTO);
        if (!StringUtils.isEmpty(baseDataSource.getPassword())) {
            baseDataSource.setPassword(DESUtils.encrypt(baseDataSource.getPassword()));
        }
        String connectionParams = JSON.toJSONString(baseDataSource);
        dataSourceDTO.setConnectionParams(connectionParams);
        dataSourceDTO.setDatasourceStatus(1);

        dataSourceService.insertSelective(dataSourceDTO);
    }

    private String checkProcessNodeJson(ProcessDefinition processDefinition) {
        String processDefinitionJson = processDefinition.getProcessDefinitionJson();
        ProcessData processData = JSONUtils.parseObject(processDefinitionJson, ProcessData.class);
        if (processData != null) {
            checkProcessNodeList(processData);
            logger.info("校验流程json成功");
        }
        return processDefinitionJson;
    }

    /**
     * update process definition
     *
     * @param processDefinition process Definition
     */
    public boolean updateProcessDefinition(ProcessDefinition processDefinition) {
        ProcessDefinition oldProcessDefine = null;
        if(StringUtils.isNoneBlank(processDefinition.getProjectId())){//projectId不为空，则是导入覆盖
            oldProcessDefine = this.mapper.selectByIdAndProjectId(processDefinition.getId(),processDefinition.getProjectId());
        }else {
            oldProcessDefine = selectByPrimaryKey(processDefinition.getId());
        }
        if (oldProcessDefine == null) {
            //            // check process definition exists
            //            logger.error("process definition {} does not exist", processDefinition.getId());
            //            throw new BusinessException(String.format(
            //                    "process definition %s does not exist", processDefinition.getId()));
            /** 导入覆盖时，系统不存在，直接新增 */
            createProcessDefinition(processDefinition);
            return true;
        } else if (oldProcessDefine.getReleaseState() == ReleaseState.ONLINE.getCode()) {
            // online can not permit edit
            logger.error("process definition {} does not allow edit", processDefinition);
            throw new BusinessException("Process Definition does not allow edit.");
        }

        // 检查JSON串合法性
        String processDefinitionJson = checkProcessNodeJson(processDefinition);

        if (processDefinition.getReleaseState() == null) {
            processDefinition.setReleaseState(ReleaseState.OFFLINE.getCode());
        }

        // 记录日志
        String logTitle = "";
        if (processDefinition.getName() != null
                && !processDefinition.getName().equals(oldProcessDefine.getName())) {
            logTitle =
                    "修改流程名称:旧名称【" + oldProcessDefine.getName() + "}，新名称【" + processDefinition.getName() + "】";
        }
        if (processDefinition.getProcessDefinitionJson() != null
                && !processDefinition
                .getProcessDefinitionJson()
                .equals(oldProcessDefine.getProcessDefinitionJson())) {
            logTitle += "修改流程设计:【" + oldProcessDefine.getName() + "】";
        }
        BdpLogUtil.log("数据计算模块", logTitle, PlatformConstant.OpType.update);

        processDefinition.setType(oldProcessDefine.getType());
        processDefinition.setProcessDefinitionJson(processDefinitionJson);
        processDefinition.setProjectId(oldProcessDefine.getProjectId());
        // 解析数据血缘
        handleLineageRelation(processDefinition);
        updateByPrimaryKeySelective(processDefinition);

        // 新增到版本表
        ProcessDefinitionHistory processDefinitionHistory = new ProcessDefinitionHistory();
        processDefinitionHistory.setId(ComUtil.getId());
        processDefinitionHistory.setProcessDefinitionId(processDefinition.getId());
        Integer versionNo =
                processDefinitionHistoryService.getNewVersionNo(processDefinition.getId()) == null
                        ? 0
                        : processDefinitionHistoryService.getNewVersionNo(processDefinition.getId());
        Integer definitionVersionNo = versionNo + 1;
        String dateFormat = DateUtils.getCurrentTime("yyyy-MM-dd HH:mm:ss");
        String definitionVersion = "版本" + definitionVersionNo.toString() + "  " + dateFormat;
        processDefinitionHistory.setDefinitionVersion(definitionVersion);
        processDefinitionHistory.setDefinitionVersionNo(definitionVersionNo);
        processDefinitionHistory.setProcessDefinitionJson(processDefinitionJson);
        processDefinitionHistory.setLocations(processDefinition.getLocations());
        processDefinitionHistory.setConnects(processDefinition.getConnects());
        processDefinitionHistoryService.insert(processDefinitionHistory);

        return true;
    }

    /**
     * release process definition: online / offline
     *
     * @param processDefinitionId process definition Id
     * @param releaseState        1-启用 0-停用
     */
    @Transactional(rollbackFor = RuntimeException.class)
    public boolean releaseProcessDefinition(String processDefinitionId, int releaseState) {
        // check state
        ReleaseState state = ReleaseState.getEnum(releaseState);
        if (state == null) {
            logger.error("request parameter {} is not valid", releaseState);
            throw new BusinessException(
                    MessageFormat.format("request parameter {0} is not valid", releaseState));
        }

        ProcessDefinition processDefinition = selectByPrimaryKey(processDefinitionId);
        if (processDefinition == null) {
            logger.error("process Definition {} doesn't exist.", processDefinitionId);
            throw new BusinessException("process Definition is not exist");
        }

        List<Schedule> scheduleList =
                schedulerService.queryByProcessDefinitionId(processDefinition.getId());
        if (scheduleList.isEmpty()) {
            logger.error("{} 定时管理未设置", processDefinition.getName());
            throw new BusinessException(MessageFormat.format("{0} 定时管理未设置", processDefinition.getName()));
        }

        switch (state) {
            case ONLINE:
                processDefinition.setReleaseState(state.getCode());
                updateByPrimaryKey(processDefinition);
                // check sub process definition release state
                List<String> subProcessDefineIds = new ArrayList<>();
                processService.recurseFindSubProcessId(processDefinitionId, subProcessDefineIds);
                String[] idArray = subProcessDefineIds.toArray(new String[0]);
                if (subProcessDefineIds.size() > 0) {
                    List<ProcessDefinition> subProcessDefinitionList =
                            mapper.queryDefinitionListByIdList(idArray);
                    if (subProcessDefinitionList != null && subProcessDefinitionList.size() > 0) {
                        for (ProcessDefinition subProcessDefinition : subProcessDefinitionList) {
                            // if there is no online process, exit directly
                            if (subProcessDefinition.getReleaseState() != ReleaseState.ONLINE.getCode()) {
                                logger.info(
                                        "not release process definition id: {} , name : {}",
                                        subProcessDefinition.getId(),
                                        subProcessDefinition.getName());
                                throw new BusinessException(
                                        MessageFormat.format(
                                                Status.PROCESS_DEFINE_NOT_RELEASE.getMsg(),
                                                subProcessDefinition.getName()));
                            }
                        }
                    }
                }
                // check master server exists
                List<Server> masterServers = monitorService.getServerListFromZK(true);
                if (masterServers.size() == 0) {
                    throw new BusinessException(Status.MASTER_NOT_EXISTS.getMsg());
                }
                // 如果是目录监控节点，则单独处理调度
                if (processDefinition
                        .getProcessDefinitionJson()
                        .contains(TaskType.MONITORDIRECTORY.name())) {
                    Schedule schedule = scheduleList.get(0);
                    // 在流程启用处，需要增加定时扫描目录
                    String jobName = QuartzExecutors.buildJobName(processDefinition.getId());
                    String jobGroupName = QuartzExecutors.buildJobGroupName(processDefinition.getProjectId());

                    Map<String, Object> dataMap = new HashMap<>(3);
                    dataMap.put(Constants.PROJECT_ID, processDefinition.getProjectId());
                    dataMap.put(Constants.PROCESS_DEFINITION_ID, processDefinition.getId());
                    dataMap.put(Constants.SCHEDULE_ID, schedule.getId());
                    // 调度时间也是有流程调度控制
                    QuartzExecutors.getInstance()
                            .addJob(
                                    MonitorDirectoryScheduleJob.class,
                                    jobName,
                                    jobGroupName,
                                    schedule.getStartTime(),
                                    schedule.getEndTime(),
                                    schedule.getCrontab(),
                                    dataMap);

                } else {
                    // 启动调度
                    for (Schedule schedule : scheduleList) {
                        setSchedule(processDefinition.getProjectId(), schedule.getId());
                    }
                }

                break;
            case OFFLINE:
                processDefinition.setReleaseState(state.getCode());
                updateByPrimaryKey(processDefinition);
                // 停止调度
                for (Schedule schedule : scheduleList) {
                    deleteSchedule(processDefinition.getProjectId(), schedule.getId());
                }
                break;
            default:
                logger.error("request parameter {} is not valid", releaseState);
                throw new BusinessException(
                        MessageFormat.format("request parameter {0} is not valid", releaseState));
        }
        // 记录日志
        String logTitle = "";
        if (releaseState == 1) {
            logTitle = "发布流程：【";
        } else {
            logTitle = "撤回流程：【";
        }
        logTitle += processDefinition.getName() + "】";
        BdpLogUtil.log("数据计算模块", logTitle, PlatformConstant.OpType.update);

        return true;
    }

    /**
     * 启动调度
     *
     * @param projectId  项目ID
     * @param scheduleId 定时调度任务ID
     * @throws RuntimeException 异常
     */
    private void setSchedule(String projectId, String scheduleId) throws RuntimeException {
        logger.info("set schedule, project id: {}, scheduleId: {}", projectId, scheduleId);

        Schedule schedule = processService.querySchedule(scheduleId);
        if (schedule == null) {
            logger.warn("process schedule info not exists");
            return;
        }

        Date startDate = schedule.getStartTime();
        Date endDate = schedule.getEndTime();

        String jobName = QuartzExecutors.buildJobName(scheduleId);
        String jobGroupName = QuartzExecutors.buildJobGroupName(projectId);

        Map<String, Object> dataMap = QuartzExecutors.buildDataMap(projectId, scheduleId, schedule);

        QuartzExecutors.getInstance()
                .addJob(
                        ProcessScheduleJob.class,
                        jobName,
                        jobGroupName,
                        startDate,
                        endDate,
                        schedule.getCrontab(),
                        dataMap);
    }

    /**
     * delete schedule
     *
     * @param projectId  project id
     * @param scheduleId schedule id
     * @throws RuntimeException runtime exception
     */
    private void deleteSchedule(String projectId, String scheduleId) throws RuntimeException {
        logger.info("delete schedules of project id:{}, schedule id:{}", projectId, scheduleId);

        String jobName = QuartzExecutors.buildJobName(scheduleId);
        String jobGroupName = QuartzExecutors.buildJobGroupName(projectId);

        if (!QuartzExecutors.getInstance().deleteJob(jobName, jobGroupName)) {
            logger.warn("set offline failure:projectId:{}, scheduleId:{}", projectId, scheduleId);
            throw new RuntimeException("set offline failure");
        }
    }

    /**
     * delete process definition by id
     *
     * @param processDefinitionId process Definition Id
     */
    @Transactional(rollbackFor = RuntimeException.class)
    public boolean deleteProcessDefinitionById(String processDefinitionId) {
        ProcessDefinition processDefinition = selectByPrimaryKey(processDefinitionId);
        if (processDefinition == null) {
            logger.error("process definition {} does not exist", processDefinitionId);
            throw new BusinessException("process Definition does not exist");
        }

        // check process definition is already online
        if (processDefinition.getReleaseState() == ReleaseState.ONLINE.getCode()) {
            logger.error("process definition {} is already on line", processDefinitionId);
            throw new BusinessException(
                    "process definition " + processDefinitionId + " is already on line");
        }

        // get the timing according to the process definition
        List<Schedule> schedules = schedulerService.queryByProcessDefinitionId(processDefinitionId);
        if (!schedules.isEmpty() && schedules.size() > 1) {
            logger.warn("scheduler num is {},Greater than 1", schedules.size());
            throw new BusinessException(
                    MessageFormat.format("scheduler num is {0},Greater than 1", schedules.size()));
        } else if (schedules.size() == 1) {
            Schedule schedule = schedules.get(0);
            schedulerService.deleteScheduleById(schedule.getId());
        }

        processResourceConfigService.deleteByProcessDefinitionId(processDefinitionId);

        deleteProcessById(processDefinitionId);
        return true;
    }

    public List<ProcessDefinition> queryDefinitionListByIdList(String[] idArray) {
        return this.mapper.queryDefinitionListByIdList(idArray);
    }

    /**
     * 根据id删除流程
     *
     * @param id 流程定义ID
     */
    @Transactional(rollbackFor = Exception.class)
    public void deleteProcessById(String id) {
        if (StringUtils.isBlank(id)) {
            throw new BusinessException("删除失败,传入的参数主键为空!");
        }
        try {
            // 记录日志
            ProcessDefinition processDefinition = selectByPrimaryKey(id);
            if (processDefinition == null) {
                throw new BusinessException("删除失败,流程定义不存在!");
            }

            // 删除任务节点实例
            TaskInstance taskInstance = new TaskInstance();
            taskInstance.setProcessDefinitionId(id);
            taskInstanceService.delete(taskInstance);

            // 如果是增量同步任务，删除缓存时间
            redisCacheHelper.del(
                    avicit.bdp.core.constant.Constants.REDIS_KEY_INCREMENT_LAST_SYNC_TIME
                            + taskInstance.getName());

            // 删除流程实例
            ProcessInstance processInstance = new ProcessInstance();
            processInstance.setProcessDefinitionId(id);
            processInstanceMapper.delete(processInstance);

            // 删除关联血缘
            LineageTableDTO lineageTable = new LineageTableDTO();
            lineageTable.setProcessDefinitionId(processDefinition.getId());
            lineageTableService.delete(lineageTable);
            LineageFieldDTO lineageField = new LineageFieldDTO();
            lineageField.setProcessDefinitionId(processDefinition.getId());
            lineageFieldService.delete(lineageField);

            // 删除流程定义
            this.mapper.deleteByPrimaryKey(id);

            BdpLogUtil.log4Delete(processDefinition);
        } catch (Exception e) {
            throw new BusinessException(e.getMessage(), e);
        }
    }

    /**
     * query process definition list paging
     *
     * @param searchVal search value
     * @param pageNo    page number
     * @param pageSize  page size
     * @param userId    user id
     * @return process definition page
     */
    public QueryRespBean<ProcessDefinition> queryProcessDefinitionListPaging(
            String projectId,
            String projectName,
            String searchVal,
            Integer pageNo,
            Integer pageSize,
            String userId) {
        QueryRespBean<ProcessDefinition> queryRespBean = new QueryRespBean<>();
        if (StringUtils.isBlank(projectId)) {
            Project project = projectMapper.queryByName(projectName);
            if (project != null) {
                projectId = project.getId();
            }
        }
        PageHelper.startPage(pageNo, pageSize);
        Page<ProcessDefinition> processDefinitions =
                this.mapper.queryDefineListPaging(searchVal, userId, projectId, false);
        queryRespBean.setResult(processDefinitions);
        BdpLogUtil.log4Query(queryRespBean);
        return queryRespBean;
    }

    /**
     * 分页查询数据开发列表
     *
     * @param typeId   分类ID
     * @param pageNo   page number
     * @param pageSize page size
     * @param keyWords 关键字
     * @return process definition page
     */
    public QueryRespBean<ProcessDefinitionDTO> getDataList(
            String typeId,
            String projectId,
            Integer type,
            String keyWords,
            Integer status,
            Integer pageNo,
            Integer pageSize) {
        QueryRespBean<ProcessDefinitionDTO> queryRespBean = new QueryRespBean<>();

        List<String> typeIds =
                commonTypeService.getChildrenIds(CommonTypeEnums.DATA_DEV, typeId, projectId);
        if (typeIds == null) {
            typeIds = new ArrayList<>();
        }
        typeIds.add(typeId);

        Map<String, Object> instanceMap = Maps.newHashMap();
        List<String> definitionIdList = Lists.newArrayList();
        PageHelper.startPage(pageNo, pageSize);
        Page<ProcessDefinitionDTO> processDefinitions =
                this.mapper.getDataListPaging(typeIds, projectId, type, status, keyWords);

        for (ProcessDefinitionDTO processDefinition : processDefinitions) {
            definitionIdList.add(processDefinition.getId());
        }
        List<ProcessInstance> processInstanceList = null;
        if (CollectionUtils.isNotEmpty(definitionIdList)) {
            processInstanceList = processInstanceMapper.queryByProcessDefineIds(definitionIdList);
        }
        if (CollectionUtils.isNotEmpty(processInstanceList)) {
            for (ProcessInstance processInstance : processInstanceList) {
                instanceMap.put(processInstance.getProcessDefinitionId(), processInstance);
            }
        }

        for (ProcessDefinitionDTO processDefinition : processDefinitions) {

            ProcessInstance processInstance =
                    (ProcessInstance) instanceMap.get(processDefinition.getId());
            if (processInstance != null) {
                ExecutionStatus executionStatus = ExecutionStatus.of(processInstance.getState());
                processDefinition.setRunningState(executionStatus.getCode());
                processDefinition.setRunningStateName(executionStatus.getDescp());
                processDefinition.setStartTime(processInstance.getStartTime());
                processDefinition.setEndTime(processInstance.getEndTime());
                if (executionStatus.typeIsRunning()) {
                    try {
                        List<TaskInstance> taskInstances =
                                taskInstanceService.queryTaskListByProcessId(processInstance.getId());
                        if (CollectionUtils.isNotEmpty(taskInstances)) {
                            int totalTaskNum = taskInstances.size();
                            List<TaskInstanceDTO> runningTaskList = new LinkedList<>();
                            for (TaskInstance taskInstance : taskInstances) {
                                if (taskInstance.getState() == ExecutionStatus.RUNNING_EXEUTION.getCode()) {
                                    TaskInstanceDTO runningTask = new TaskInstanceDTO();
                                    runningTask.setTaskId(taskInstance.getId());
                                    runningTask.setTaskName(taskInstance.getName());
                                    runningTask.setProcessStartTime(processDefinition.getStartTime());
                                    runningTaskList.add(runningTask);
                                }
                            }
                            processDefinition.setTaskList(runningTaskList);
                            int runningTaskNum = runningTaskList.size();
                            processDefinition.setNoteRate(runningTaskNum + "/" + totalTaskNum);
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

            } else {
                processDefinition.setRunningStateName("未运行");
            }
        }

        valueConvertList(processDefinitions);
        queryRespBean.setResult(processDefinitions);
        BdpLogUtil.log4Query("数据开发模块", "数据开发", queryRespBean);
        return queryRespBean;
    }

    private void valueConvertList(Page<ProcessDefinitionDTO> processDefinitions) {
        // 循环组装请求数据
        Map<String, Set<String>> convertFormData = new HashMap<>(16);
        for (ProcessDefinitionDTO processDefinitionDTO : processDefinitions) {
            BusinessUtil.createConvertSet(
                    convertFormData, SystemConstant.USER, processDefinitionDTO.getCreatedBy());
        }
        if (convertFormData.size() > 0) {
            // 获取请求结果
            Map<String, Map<String, String>> convertResultData =
                    convertColumnClient.replace(convertFormData);
            // 循环设置Alias或Name的值
            for (ProcessDefinitionDTO processDefinitionDTO : processDefinitions) {
                processDefinitionDTO.setCreatedUserName(
                        BusinessUtil.convertFormat(
                                convertResultData, SystemConstant.USER, processDefinitionDTO.getCreatedBy()));
            }
        }
    }

    /**
     * 获取流程定义信息
     *
     * @param id 流程定义id
     * @return 流程定义
     */
    public ProcessDefinition getProcessDefinition(String id) {
        ProcessDefinition processDefinition = this.selectByPrimaryKey(id);
        valueConvert(processDefinition);
        BdpLogUtil.log4Query(processDefinition);
        return processDefinition;
    }

    private void valueConvert(ProcessDefinition processDefinition) {
        // 循环组装请求数据
        try {
            Map<String, Set<String>> convertFormData = new HashMap<>(16);
            BusinessUtil.createConvertSet(
                    convertFormData, SystemConstant.USER, processDefinition.getCreatedBy());
            if (convertFormData.size() > 0) {
                // 获取请求结果
                Map<String, Map<String, String>> convertResultData =
                        convertColumnClient.replace(convertFormData);
                // 循环设置Alias或Name的值
                processDefinition.setCreatedUserName(
                        BusinessUtil.convertFormat(
                                convertResultData, SystemConstant.USER, processDefinition.getCreatedBy()));
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * 获取流程定义数量
     *
     * @param projectIdList 工作空间集合
     * @return 流程定义数量
     */
    public Long getDefinitionByProjectIds(List<String> projectIdList) {
        return this.mapper.getDefinitionByProjectIds(projectIdList);
    }

    public Long getDefinitionByProjectId(String projectId) {
        return this.mapper.getDefinitionByProjectId(projectId);
    }

    public Boolean verifyProcessDefinitionName(
            String projectId, String typeId, String name, Integer type, String id) {
        if (StringUtils.isNotBlank(id)) {
            ProcessDefinition processDefinition = mapper.queryByDefineId(id);
            if (processDefinition == null) {
                throw new BusinessException("流程实例不存在");
            } else {
                if (name.equals(processDefinition.getName())) {
                    return true;
                }
            }
        }
        Integer size = this.mapper.countDefinitionByName(projectId, typeId, name, type);
        return size <= 0;
    }

    public List<CommonTypeDTO> getTypeAndProcessDefinitionList(
            String projectId, Integer type, String processDefinitionId) {
    /* List<CommonTypeDTO> result = bdpDasCommonTypeService.findCommonType(CommonTypeEnums.DATA_DEV, projectId);
    if (CollectionUtils.isNotEmpty(result)) {
        List<ProcessDefinition> processDefinitions = this.mapper.queryAllDefinitionList(projectId);
        if (CollectionUtils.isNotEmpty(processDefinitions)) {
            Map<String, List<ProcessDefinition>> map = new HashMap<>(16);
            processDefinitions.stream()
                    .filter(item -> !item.getId().equals(processDefinitionId))
                    .collect(Collectors.groupingBy(ProcessDefinition::getTypeId, Collectors.toList()))
                    .forEach(map::put);
            for (CommonTypeDTO commonTypeDTO : result) {
                handleTypeWithProcessDefinition(map, commonTypeDTO);
            }
        }
    }*/
        return this.getDataDevList(0, projectId, type, "", processDefinitionId);
    }

    private void handleTypeWithProcessDefinition(
            Map<String, List<ProcessDefinition>> map, CommonTypeDTO commonTypeDTO) {
        if (map.containsKey(commonTypeDTO.getId())) {
            List<ProcessDefinitionDTO> dtoList =
                    map.get(commonTypeDTO.getId()).stream()
                            .map(
                                    processDefinition -> {
                                        ProcessDefinitionDTO dto = new ProcessDefinitionDTO();
                                        dto.setId(processDefinition.getId());
                                        dto.setName(processDefinition.getName());
                                        return dto;
                                    })
                            .collect(Collectors.toList());
            commonTypeDTO.setProcessDefinitionList(dtoList);
        }
        List<CommonTypeDTO> children = commonTypeDTO.getChildren();
        if (CollectionUtils.isNotEmpty(children)) {
            for (CommonTypeDTO child : children) {
                handleTypeWithProcessDefinition(map, child);
            }
        }
    }

    /**
     * 根据类型或者项目空间统计流程任务
     *
     * @param filed 统计字段 【type, projectId】
     * @return 流程任务统计列表
     */
    public List<ProcessDefinitionStateDto> bigScreen(String filed) {
        String userId = ThreadContextHelper.getUserId();
        if (StringUtils.isBlank(userId)) {
            userId = "1";
        }

        if ("type".equals(filed)) {
            return mapper.countDefinitionByType(userId);
        } else if ("projectId".equals(filed)) {
            return mapper.countDefinitionByProjectId(userId);
        } else {
            return null;
        }
    }

    /**
     * 根据项目ID获取流程列表
     *
     * @param projectId
     * @return java.util.List<avicit.bdp.dds.dao.entity.ProcessDefinition>
     * @author xugb
     * @date 2021/1/18 13:54
     */
    public List<ProcessDefinition> getAllProcessDefinitionList(
            String projectId, String processDefinitionId) {
        ProcessDefinition query = new ProcessDefinition();
        query.setProjectId(projectId);
        query.setType(6);//数据开发工作台
        List<ProcessDefinition> processDefinitionList = this.selectList(query);
        List<ProcessDefinition> list = new ArrayList<>();
        for (ProcessDefinition definition : processDefinitionList) {
            // 排除自己本身的流程
            if (StringUtils.isNotBlank(processDefinitionId)
                    && definition.getId().equals(processDefinitionId)) {
                continue;
            }
            if (definition.getProcessDefinitionJson() != null) {
                ProcessData processData =
                        JSONUtils.parseObject(definition.getProcessDefinitionJson(), ProcessData.class);
                if (processData == null) {
                    continue;
                }
                List<TaskNode> taskNodeList =
                        (processData.getTasks() == null) ? new ArrayList<>() : processData.getTasks();
                if (!taskNodeList.isEmpty()) {
                    list.add(definition);
                }
            }
        }
        return list;
    }

    /**
     * 根据流程定义id获取节点信息
     *
     * @param processDefinitionId
     * @return java.util.List<avicit.bdp.common.dispatch.model.TaskNode>
     * @author xugb
     * @date 2021/1/18 14:05
     */
    public List<TaskNode> getAllTaskNodeist(String processDefinitionId) {
        ProcessDefinition processDefinition = this.selectByPrimaryKey(processDefinitionId);
        if (processDefinition == null) {
            logger.info("流程定义不存在");
            throw new BusinessException(("流程定义不存在"));
        }
        String processDefinitionJson = processDefinition.getProcessDefinitionJson();

        ProcessData processData = JSONUtils.parseObject(processDefinitionJson, ProcessData.class);

        // process data check
        if (null == processData) {
            logger.error("process data is null");
            return new ArrayList<>();
        }

        List<TaskNode> taskNodeList =
                (processData.getTasks() == null) ? new ArrayList<>() : processData.getTasks();
        return taskNodeList;
    }

    public Map<String, List<TaskNode>> getTaskNodeListByDefinitionIdList(
            String processDefinitionIdList) {
        Map<String, List<TaskNode>> taskNodeMap = new HashMap<>(16);
        String[] idList = processDefinitionIdList.split(",");
        List<String> idIntList = new ArrayList<>();
        for (String definitionId : idList) {
            idIntList.add(definitionId);
        }
        String[] idArray = idIntList.toArray(new String[idIntList.size()]);
        List<ProcessDefinition> processDefinitionList =
                this.mapper.queryDefinitionListByIdList(idArray);
        if (CollectionUtils.isEmpty(processDefinitionList)) {
            logger.info("process definition not exists");
            return taskNodeMap;
        }

        for (ProcessDefinition processDefinition : processDefinitionList) {
            String processDefinitionJson = processDefinition.getProcessDefinitionJson();
            ProcessData processData = JSONUtils.parseObject(processDefinitionJson, ProcessData.class);
            if (processData == null) {
                continue;
            }
            List<TaskNode> taskNodeList =
                    (processData.getTasks() == null) ? new ArrayList<>() : processData.getTasks();
            taskNodeMap.put(processDefinition.getId(), taskNodeList);
        }
        return taskNodeMap;
    }

    public List<CommonTypeDTO> getDataDevList(
            Integer releaseState,
            String projectId,
            Integer type,
            String keyWords,
            String processDefinitionId) {
        if (type == null) {
            throw new BusinessException("type为null，请检查参数！");
        }
    /*if (releaseState == null) {
        releaseState = 0;
    }*/

        CommonTypeDTO query = new CommonTypeDTO();
        if (type == 7) { // 离线计算
            query.setType(CommonTypeEnums.SPARK_FLOW);
        } else if (type == 8) { // 实时计算
            query.setType(CommonTypeEnums.SPARK_STREAM_FLOW);
        } else if (type == 9) {
            query.setType(CommonTypeEnums.SPARK_ML_FLOW);
        } else {
            query.setType(CommonTypeEnums.DATA_DEV);
        }
        query.setProjectId(projectId);
        List<CommonTypeDTO> dataList = commonTypeService.selectList(query);
        for (CommonTypeDTO dto : dataList) {
            dto.setTypeFlag(0); // 分类
            dto.setIcon("folder");
        }
        List<CommonTypeDTO> dataDevList =
                this.mapper.getDataDevList(releaseState, projectId, type, keyWords);
        userValueConvertList(dataDevList);
        if (CollectionUtils.isNotEmpty(dataDevList)) {
            for (int i = 0; i < dataDevList.size(); i++) {
                CommonTypeDTO dto = dataDevList.get(i);
                /*********************处理子任务查询*********************/
                if (StringUtils.isNotBlank(processDefinitionId)) {
                    dto.setSelectable(true);
                    if (processDefinitionId.equals(dto.getId())) {
                        dataDevList.remove(dto);
                        i--;
                        continue;
                    }
                }
                /******************************************************/
                dto.setTypeFlag(1); // 任务
                dto.setIcon("codepen");
                dto.setLeaf(1);
            }
        }
        // 处理查询或是已发布任务的分类显示
        if (StringUtils.isNotBlank(keyWords) || (releaseState != null && releaseState == 1)) {
            dataList = handleTypeListByData(dataList, dataDevList);
        }

        Collections.sort(
                dataList,
                (u1, u2) -> {
                    if (u2.getOrderBy() != null && u1.getOrderBy() != null) {
                        return u1.getOrderBy() - u2.getOrderBy();
                    } else {
                        return 0;
                    }
                });

        dataList.addAll(dataDevList);

        // 顶级节点
        List<CommonTypeDTO> list = Lists.newArrayList();
        // 子节点数据
        Map<String, List<CommonTypeDTO>> map = Maps.newHashMap();
        if (CollectionUtils.isNotEmpty(dataList)) {
            for (CommonTypeDTO commonTypeDTO : dataList) {
                if ("-1".equals(commonTypeDTO.getParentId())) {
                    list.add(commonTypeDTO);
                } else {
                    if (map.containsKey(commonTypeDTO.getParentId())) {
                        map.get(commonTypeDTO.getParentId()).add(commonTypeDTO);
                    } else {
                        List<CommonTypeDTO> commonTypeDTOList = Lists.newArrayList();
                        commonTypeDTOList.add(commonTypeDTO);
                        map.put(commonTypeDTO.getParentId(), commonTypeDTOList);
                    }
                }
            }
        }

        /******************兼容数据开发工作台**********************/
        if (StringUtils.isNotBlank(processDefinitionId)) {
            Project project = projectMapper.selectByPrimaryKey(projectId);
            if ("0".equals(project.getProjectType())) {
                CommonTypeDTO topCommonType =
                        commonTypeService.selectByPrimaryKey("402880a4735a767202735a8aeae30004");
                list.clear();
                list.add(topCommonType);
            }
        }

        // 组成树类型
        listConvert(list, map);

        return list;
    }

    private void userValueConvertList(List<CommonTypeDTO> commonTypeDTOList) {
        try {
            // 循环组装请求数据
            Map<String, Set<String>> convertFormData = new HashMap<>(16);
            for (CommonTypeDTO commonTypeDTO : commonTypeDTOList) {
                BusinessUtil.createConvertSet(
                        convertFormData, SystemConstant.USER, commonTypeDTO.getCreatedBy());
            }
            if (convertFormData.size() > 0) {
                // 获取请求结果
                Map<String, Map<String, String>> convertResultData =
                        convertColumnClient.replace(convertFormData);
                // 循环设置Alias或Name的值
                for (CommonTypeDTO commonTypeDTO : commonTypeDTOList) {
                    commonTypeDTO.setCreatedBy(
                            BusinessUtil.convertFormat(
                                    convertResultData, SystemConstant.USER, commonTypeDTO.getCreatedBy()));
                }
            }
        } catch (Exception e) {
            logger.info(e.getMessage());
        }
    }

    /**
     * 处理查询时某些树节点不显示
     *
     * @param typeList
     * @param dataDevList
     * @return java.util.List<avicit.bdp.common.service.dto.CommonTypeDTO>
     */
    private List<CommonTypeDTO> handleTypeListByData(
            List<CommonTypeDTO> typeList, List<CommonTypeDTO> dataDevList) {
        Map<String, String> map = new HashMap<>(16);
        for (CommonTypeDTO data : dataDevList) {
            if (data.getParentId() != null
                    && !"-1".equals(data.getParentId())
                    && !map.containsKey(data.getParentId())) {
                map.put(data.getParentId(), data.getParentId());
                compareTypeList(data.getParentId(), map, typeList);
            }
        }
        List<CommonTypeDTO> list = new ArrayList<>();
        for (CommonTypeDTO type : typeList) {
            if (map.containsKey(type.getId())) {
                list.add(type);
            }
        }
        return list;
    }

    private void compareTypeList(
            String typeId, Map<String, String> map, List<CommonTypeDTO> typeList) {
        for (CommonTypeDTO type : typeList) {
            if (type.getId().equals(typeId)) {
                if (!"-1".equals(type.getParentId())) {
                    map.put(type.getParentId(), type.getParentId());
                    compareTypeList(type.getParentId(), map, typeList);
                } else {
                    break;
                }
            }
        }
    }

    private void listConvert(List<CommonTypeDTO> list, Map<String, List<CommonTypeDTO>> map) {
        for (CommonTypeDTO commonTypeDTO : list) {
            List<CommonTypeDTO> commonTypeList = map.get(commonTypeDTO.getId());
            if (CollectionUtils.isNotEmpty(commonTypeList)) {
                listConvert(commonTypeList, map);
                commonTypeDTO.setChildren(commonTypeList);
            } else {
                commonTypeDTO.setChildren(null);
            }
        }
    }

    public List<CommonTypeDTO> getDataDevTypeTree(String projectId, String type) {
        CommonTypeDTO query = new CommonTypeDTO();
        query.setType(type);
        query.setProjectId(projectId);
        List<CommonTypeDTO> dataList = commonTypeService.selectList(query);
        for (CommonTypeDTO dto : dataList) {
            dto.setTypeFlag(0); // 分类
        }

        Collections.sort(dataList, (u1, u2) -> u2.getCreationDate().compareTo(u1.getCreationDate()));

        // 顶级节点
        List<CommonTypeDTO> list = Lists.newArrayList();
        // 子节点数据
        Map<String, List<CommonTypeDTO>> map = Maps.newHashMap();
        if (CollectionUtils.isNotEmpty(dataList)) {
            for (CommonTypeDTO commonTypeDTO : dataList) {
                if ("-1".equals(commonTypeDTO.getParentId())) {
                    list.add(commonTypeDTO);
                } else {
                    if (map.containsKey(commonTypeDTO.getParentId())) {
                        map.get(commonTypeDTO.getParentId()).add(commonTypeDTO);
                    } else {
                        List<CommonTypeDTO> commonTypeDTOList = Lists.newArrayList();
                        commonTypeDTOList.add(commonTypeDTO);
                        map.put(commonTypeDTO.getParentId(), commonTypeDTOList);
                    }
                }
            }
        }
        // 组成树类型
        listConvert(list, map);

        return list;
    }

    public QueryRespBean<ProcessDefinitionHistory> getDefinitionHistoryList(
            String processDefinitionId, Integer pageNo, Integer pageSize) {
        QueryRespBean<ProcessDefinitionHistory> queryRespBean = new QueryRespBean<>();
        if (StringUtils.isBlank(processDefinitionId)) {
            throw new BusinessException("请检查参数");
        }
        PageHelper.startPage(pageNo, pageSize);
        Page<ProcessDefinitionHistory> processDefinitionsHistory =
                processDefinitionHistoryService.getDefinitionHistoryList(processDefinitionId);
        queryRespBean.setResult(processDefinitionsHistory);
        BdpLogUtil.log4Query(queryRespBean);
        return queryRespBean;
    }

    public Boolean deleteProcessDefinitionHistory(String processDefinitionId) {
        processDefinitionHistoryService.deleteProcessDefinitionHistory(processDefinitionId);
        return true;
    }

    /**
     * 流程定义导出
     *
     * @param id
     * @param response
     * @return
     */
    public boolean exportFlow(String id, HttpServletResponse response) {
        if (StringUtils.isBlank(id)) {
            throw new BusinessException("流程定义ID为空.");
        }

        ProcessDefinition queryRet = this.getProcessDefinition(id);
        if (queryRet == null) {
            throw new BusinessException("未找到流程定义,id=." + id);
        }

        exportFlowInner(queryRet, response);

        return true;
    }

    /**
     * 流程定义导入(返回文件中对象)
     *
     * @param file
     * @return
     */
    public ProcessDefinition importFlow(MultipartFile file) {
        ProcessDefinition entity = new ProcessDefinition();
        try {

            entity = (ProcessDefinition) XmlUtils.fromXml(file.getInputStream(), entity);
            //状态改为未发布
            entity.setReleaseState(0);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return entity;
    }

    /**
     * 流程定义导入
     *
     * @param processDefinition
     * @param response
     */
    public void exportFlowInner(ProcessDefinition processDefinition, HttpServletResponse response) {

        try {
            /*************************************序列化为xml*************************************/

            ByteArrayOutputStream byteOutput = new ByteArrayOutputStream();
            OutputStream outputStream = response.getOutputStream();
            String name = URLEncoder.encode(processDefinition.getName(), "utf-8");

            // 将Java对象序列化成XML
            // ObjectOutputStream oos = new ObjectOutputStream(byteOutput);
            PrintWriter writer = new PrintWriter(byteOutput);
            XmlUtils.toXml(processDefinition, writer);
            response.setContentType("application/xml; charset=utf-8");
            response.setHeader("Content-Disposition", "attachment;filename=" + name + ".xml");
            response.setHeader("Pragma", "No-cache");
            response.setHeader("Expires", "0");
            byteOutput.writeTo(outputStream);
            byteOutput.flush();
            outputStream.flush();
            byteOutput.close();
            writer.close();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    public List<ProcessDefinition> getDcsProcess(String projectId, Integer type) {
        ProcessDefinition query = new ProcessDefinition();
        query.setType(type);
        query.setProjectId(projectId);
        return this.selectList(query);
    }

    /**
     * 根据标准sql获取字段名
     *
     * @param sql
     * @return java.util.List<java.lang.String>
     * @author felix
     * @date 2022/6/14 10:30
     */
    public List<String> getCloumnNamesBysQL(String sql) {
        List<String> columnList = new ArrayList<>();
        String dbType = DbType.MYSQL.getDescp(); // 以mysql为标准
        try {
            List<SQLStatement> stmtList = null;
            try {
                stmtList = SQLUtils.parseStatements(sql, dbType);
            } catch (ParserException e) {
                throw new BusinessException("sql语法有误，请检查sql:" + e.getMessage());
            }
            if (CollectionUtils.isEmpty(stmtList)) {
                return null;
            }
            // 解析sql，获取表名
            for (SQLStatement sqlStatement : stmtList) {
                // mysql为例，若为其他数据源请换成对应的SchemaStatVisitor
                MySqlSchemaStatVisitor visitor = new MySqlSchemaStatVisitor();
                // OdpsSchemaStatVisitor visitor = new OdpsSchemaStatVisitor();
                sqlStatement.accept(visitor);

                // 获取查询字段
                Collection<TableStat.Column> columns = visitor.getColumns();
                Map<String, String> map = new HashMap<>();
                if (CollectionUtils.isNotEmpty(columns)) {
                    for (TableStat.Column column : columns) {
                        if ("*".equals(column.getName())) {
                            continue;
                        }
                        if (!map.containsKey(column.getName())) {
                            columnList.add(column.getName());
                            map.put(column.getName(), column.getName());
                        }
                    }
                }
            }
            return columnList;
        } catch (Exception e) {
            throw new BusinessException("**************异常SQL*****************" + e.getMessage());
        }
    }

    @Transactional
    public String createAndRunProcessDefinition(ProcessDefinition processDefinition)
            throws ParseException {
        // 检查流程定义是否存在
        if (StringUtils.isBlank(processDefinition.getProcessDefinitionJson())) {
            throw new BusinessException("流程节点尚未定义，请先设计流程");
        }
        String processDefinitionId = this.createProcessDefinition(processDefinition);
        //        if (processDefinition.getReleaseState().intValue() == ReleaseState.OFFLINE.getCode())
        // {
        //            this.releaseProcessDefinition(processDefinitionId, ReleaseState.ONLINE.getCode());
        //        }

        String scheduleTime = "";
        String startNodeList = "";
        Integer failureStrategy = FailureStrategy.END.getCode();
        Integer execType = CommandType.START_PROCESS.getCode();
        Integer warningType = WarningType.FAILURE.getCode();
        Integer timeout = Constants.MAX_TASK_TIMEOUT;
        String warningUserIds = "";
        String workerGroup = "";
        Integer taskDependType = TaskDependType.TASK_POST.getCode();
        Integer processInstancePriority = Priority.MEDIUM.getCode();
        Integer runMode = RunMode.RUN_MODE_SERIAL.getCode();
        // 启动流程
        executorService.execProcessInstance(
                processDefinitionId,
                scheduleTime,
                execType,
                failureStrategy,
                startNodeList,
                taskDependType,
                warningType,
                warningUserIds,
                runMode,
                processInstancePriority,
                workerGroup,
                timeout);

        // 记录日志
        String logTitle = "运行任务：【" + processDefinition.getName() + "】";
        BdpLogUtil.log("数据计算模块", logTitle, PlatformConstant.OpType.insert);
        return processDefinitionId;
    }

    @Transactional
    public String queryLastOneProcessId(String processDefinitionId) {
        String processInstanceId = "";
        ProcessInstance instance = processInstanceMapper.queryLastOneProcess(processDefinitionId);
        if (instance != null) {
            processInstanceId = instance.getId();
        }
        logger.info("任务：" + processDefinitionId + "--" + processInstanceId);
        return processInstanceId;
    }
}
